// @Author EthanScriptOn
// @Desc
package kafka_operator

import (
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/common/identity"
	"gitee.com/fatzeng/srf_switch_expansion_package/connect"
	"gitee.com/fatzeng/srf_switch_expansion_package/extend_config"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/actuator"
	"github.com/confluentinc/confluent-kafka-go/kafka"
)

type KafkaProduceOperator struct {
	*identity.IdentityInformationCarrier
}

func (k *KafkaProduceOperator) Execute(ctx *actuator.SrfSwitchBridgeActuatorContext, configIdentifier interface{}, structureParam interface{}) (interface{}, error) {
	messages, ok := structureParam.([]*kafka.Message)
	if !ok {
		return nil, fmt.Errorf("structureParam is not []*kafka.Message")
	}
	if len(messages) <= 0 {
		return nil, fmt.Errorf("messages is empty")
	}
	producer, err := connect.GetKafkaSyncProducer()
	if err != nil {
		return nil, err
	}
	responses := make([]*extend_config.KafkaOperatorResponse, 0)
	for _, message := range messages {
		err := producer.Produce(message, nil)
		responses = append(responses, &extend_config.KafkaOperatorResponse{Partition: message.TopicPartition.Partition, Offset: int64(message.TopicPartition.Offset), Err: err, OriginMsg: message})
	}
	return responses, nil
}
